package com.lagou.gkk;

import com.lagou.bak.PayEvent;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 *  乘客下单以后两分钟内没有接单
 */
public class Damo1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        //乘客打车产生数据 uid state ts
        //乘客下单，超时未接单
        DataStreamSource<Chengke> data = env.fromElements(
                new Chengke(1, "create", 1597905234000L),
                new Chengke(1, "ok", 1597905235000L),
                new Chengke(2, "create", 1597905236000L),
                new Chengke(2, "ok", 1597905237000L),
                new Chengke(3, "create", 1597905239000L),
                new Chengke(4, "create", 1597905238000L)
        );

        SingleOutputStreamOperator<Chengke> watermarks = data.assignTimestampsAndWatermarks(new WatermarkStrategy<Chengke>() {
            @Override
            public WatermarkGenerator<Chengke> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Chengke>() {
                    long maxTimestamp = Long.MIN_VALUE;

                    @Override
                    public void onEvent(Chengke event, long eventTimestamp, WatermarkOutput output) {
                        maxTimestamp = Math.max(maxTimestamp, event.getTs());
                    }

                    long maxOutOfOrderness = 500l;

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxTimestamp - maxOutOfOrderness));
                    }
                };
            }
        }.withTimestampAssigner((element, recordTimestamp) -> element.getId()));

        KeyedStream<Chengke, Integer> keyed = watermarks.keyBy(value -> value.getId());

        Pattern<Chengke, Chengke> pattern = Pattern.<Chengke>
                begin("begin").where(new IterativeCondition<Chengke>() {
            @Override
            public boolean filter(Chengke value, Context<Chengke> ctx) throws Exception {
                return value.getState().equals("create");
            }
        })
                .followedBy("ok").where(new IterativeCondition<Chengke>() {
                    @Override
                    public boolean filter(Chengke value, Context<Chengke> ctx) throws Exception {
                        return value.getState().equals("ok");
                    }
                })
                .within(Time.seconds(60*2));

        PatternStream<Chengke> patternStream = CEP.pattern(keyed, pattern);
        OutputTag<Chengke> outputTag = new OutputTag<Chengke>("outtimeok") {
        };
        SingleOutputStreamOperator<Chengke> result = patternStream.select(outputTag, new PatternTimeoutFunction<Chengke, Chengke>() {
            @Override
            public Chengke timeout(Map<String, List<Chengke>> pattern, long timeoutTimestamp) throws Exception {
                System.out.println("timeout:" + pattern);
                return pattern.get("begin").get(0);
            }
        }, new PatternSelectFunction<Chengke, Chengke>() {
            @Override
            public Chengke select(Map<String, List<Chengke>> pattern) throws Exception {
                System.out.println("select:" + pattern);
                return pattern.get("ok").get(0);
            }
        });
        DataStream<Chengke> sideOutput = result.getSideOutput(outputTag);
        sideOutput.print();
        env.execute();

    }
}
